package com.udf.flink.scala.apitest.state

import java.lang

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.metrics.Counter
import org.apache.flink.streaming.api.scala._
import scala.collection.JavaConverters._
import scala.collection.JavaConversions._

/**
 * 在listState删除指定value
 */

object FlinkListStateTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 1、从集合中读取数据
    val dataList = List("a", "b", "c", "d", "e")

    env.fromCollection(dataList)
      .setParallelism(1)
      .map(c => (c, 1))
      .keyBy(_._2)
      .map(new MyRichMapFunc)
      .print("char->")

    //执行
    env.execute("FlinkListStateTest")

  }
}

//自定义实现RichMapFunction
class MyRichMapFunc extends RichMapFunction[(String, Int), String] {
  var charState: ListState[String] = _
  var ctState: ValueState[Int] = _
  var strings: lang.Iterable[String] = _
  //metrics
  var ct: Counter = _

  override def open(parameters: Configuration): Unit = {
    //定义描述器
    val charDes = new ListStateDescriptor[String]("char", classOf[String])
    val ctDes = new ValueStateDescriptor[Int]("count", classOf[Int])
    //状态初始化
    charState = getRuntimeContext.getListState[String](charDes)
    ctState = getRuntimeContext.getState[Int](ctDes)
    //累加器初始化
    ct = getRuntimeContext.getMetricGroup.counter("ct")
  }

  override def map(character: (String, Int)): String = {
    ct.inc()
    //从状态中获取历史值
    strings = charState.get()
    val count: Int = ctState.value()
    if (strings.isEmpty && ct.getCount == 1) {
      //添加空集合
      charState.add("delete")
    }
    charState.add(character._1)
    //更新计数器
    ctState.update(count + 1)
    //将list状态转换为list集合
    val charList: List[String] = charState.get().asScala.toList
    if (ctState.value() == 3) {
      //删除delete
      val newList: List[String] = charList.dropWhile(char => {
        char.equals("delete")
      })
      //清空状态
      charState.clear()
      //更新状态
      charState.addAll(newList.asJava)
    }

    strings = charState.get()
    if (strings.nonEmpty) {
      for (x <- strings) {
        print("state->" + x + "&ct->" + ct.getCount+"   ")
      }
    }
    character._1
  }
}
